package com.zhanggm.message.message.kafka;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.bw.springboot.stater.email.bean.MailBean;
import com.bw.springboot.stater.email.service.MailService;
import com.zhanggm.message.message.entity.KafkaLog;
import com.zhanggm.message.message.service.IKafkaLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

@Component
@Slf4j
public class EmailConsumer {

    @Autowired
    private MailService mailService;

    @Autowired
    private IKafkaLogService iKafkaLogService;

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 接收消息，发送邮件
     * @param message
     */
    @KafkaListener(topics = {"hetongEmailTopic1"})
    public void consumer(String message, Acknowledgment ack) {
        log.info("接收消息：{}",message);
        MailBean mailBean = JSON.parseObject(message, MailBean.class);
        if(!redisTemplate.opsForValue().setIfAbsent(mailBean.getExtId(),null,24, TimeUnit.HOURS)){
            log.info("消息重复消费:{}",message);
            return;
        }
        //封装日志对象
        KafkaLog kafkaLog = new KafkaLog();
        kafkaLog.setMsgContent(message);
        kafkaLog.setMsgId(mailBean.getExtId());
        //0表是开始消费
        kafkaLog.setMsgStatus(0);
        kafkaLog.setMsgTime(LocalDateTime.now());
        kafkaLog.setMsgCount(1);
        try {
            //消费消息前，添加日志,防止消息丢失，消息补偿使用
            iKafkaLogService.save(kafkaLog);
            //发送邮件
            log.info("开始消费消息，mailBean：{}",mailBean);
            mailService.sendSimpleMail(mailBean);
            log.info("消费完成：{}",message);
            //手动确认消息
            ack.acknowledge();
            //消费成功修改日志状态
            kafkaLog.setMsgStatus(1);
            iKafkaLogService.saveOrUpdate(kafkaLog);
        } catch (Exception e) {
            ack.nack(1000);
            //2消费失败
            kafkaLog.setMsgStatus(2);
            iKafkaLogService.saveOrUpdate(kafkaLog);
        }
    }

}
